package vip.xls8z.learning.nio.reactor.multiple.reactor;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class Reactor implements Runnable {

    final Selector[] selectors = new Selector[2];
    final ServerSocketChannel serverSocketChannel;
    final ExecutorService executorService = Executors.newFixedThreadPool(2);

    public Reactor() throws IOException {
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();

        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8091));

        serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT, new Acceptor(selectors, serverSocketChannel));
        // 附加对象
        // selectionKey.attach(new Acceptor(selectors[1], serverSocketChannel));
    }


    /**
     * 非阻塞写法
     */
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                for (Selector selector : selectors) {
                    selectNonBlock(selector);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("run reactor error ", e);
        }
    }

    /*
     阻塞写法
     @Override
     public void run() {
         try {
             for (Selector selector : selectors) {
                 // 避免某个selector阻塞其他的selector，采用线程池处理(可以换成selector.selectNow())
                 executorService.execute(() -> {
                     while (!Thread.interrupted()) {
                         try {
                             doSelect(selector);
                         } catch (IOException e) {
                             e.printStackTrace();
                         }
                     }
                 });
             }
         } catch (Exception e) {
             throw new RuntimeException("run reactor error ", e);
         }
     }
    */

    /**
     * 阻塞
     */
    private void selectBlock(Selector selector) throws IOException {
        selector.select(); // 阻塞
        doSelect(selector);
    }

    /**
     * 非阻塞
     */
    private void selectNonBlock(Selector selector) throws IOException {
        int count = selector.selectNow();// 非阻塞
        if (count == 0) return;
        doSelect(selector);
    }

    private void doSelect(Selector selector) throws IOException {
        Set<SelectionKey> selectionKeys = selector.selectedKeys();
        Iterator<SelectionKey> iterator = selectionKeys.iterator();
        while (iterator.hasNext()) {
            SelectionKey selectionKey = iterator.next();
            dispatch(selectionKey);
        }
        selectionKeys.clear();
    }

    private void dispatch(SelectionKey selectionKey) {
        Runnable runnable = (Runnable) selectionKey.attachment();
        if (runnable != null) {
            log.info("main selector: {}, sub selector: {}, target: {}", selectors[0], selectors[1], selectionKey.selector());
            runnable.run();
        }
    }

    public static void main(String[] args) throws IOException {
        new Reactor().run();
    }
}
